package com.fluidops.datacatalog.reltables.data;

import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.log4j.Logger;
import org.openrdf.model.Literal;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.ValueFactory;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.openrdf.model.vocabulary.RDF;
import org.openrdf.query.BindingSet;
import org.openrdf.query.MalformedQueryException;
import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQuery;
import org.openrdf.query.TupleQueryResult;
import org.openrdf.repository.Repository;
import org.openrdf.repository.RepositoryConnection;
import org.openrdf.repository.RepositoryException;
import org.openrdf.repository.RepositoryResult;
import org.openrdf.repository.sail.SailRepository;
import org.openrdf.repository.sparql.SPARQLRepository;
import org.openrdf.rio.RDFFormat;
import org.openrdf.sail.memory.MemoryStore;

import com.fluidops.datacatalog.reltables.Starter;
import com.fluidops.datacatalog.reltables.data.Vocabulary.DCTERMS;
import com.fluidops.datacatalog.reltables.data.Vocabulary.DataCube;
import com.fluidops.datacatalog.reltables.data.Vocabulary.SDMX;
import com.fluidops.datacatalog.reltables.data.Vocabulary.SKOS;


public class WorldbankLabelProvider {

	public static final String CATALOG_META_URL = "http://worldbank.270a.info/data/meta/meta.rdf";
	public static final String CATALOG_NAMESPACE = "http://worldbank.270a.info/dataset/";
	public static final String DATASET_META_NAMESPACE = "http://worldbank.270a.info/classification/";

	public static final String SPARQL_ENDPOINT = "http://worldbank.270a.info/sparql";
	public static final String INDICATOR_GRAPH = "http://worldbank.270a.info/graph/world-development-indicators";
	public static final String META_GRAPH = "<http://worldbank.270a.info/graph/meta>";

	private static Logger logger = Logger.getLogger(WorldbankLabelProvider.class.getName());

	private static ValueFactory vf = ValueFactoryImpl.getInstance();

	public List<Statement> getLabelStatements() throws Exception{
		Set<Statement> labels = new HashSet<Statement>();

		InputStream is = establishLink(new URL(CATALOG_META_URL));
		Repository metaRepo = new SailRepository(new MemoryStore());
		metaRepo.initialize();

		RepositoryConnection con = metaRepo.getConnection();
		con.add(is, "", RDFFormat.RDFXML);
		con.commit();
		is.close();

		List<URI> datasetURIs = getDatasetURIs(con);
		for(URI ds : datasetURIs){
			labels.add(vf.createStatement(ds, RDF.TYPE, DataCube.DATASET));
			retrieveDatasetLabels(labels, con, ds);
		}

	//	retrieveEntities(labels);
		labels.add(vf.createStatement(vf.createURI(Starter.catalogSubject), vf.createURI(Starter.catalogProperty), vf.createURI(CATALOG_NAMESPACE)));
		logger.info("Finished loading Meta Information of \"world-development-indicators\" catalog");

		return new ArrayList<Statement>(labels);
	}

	private void retrieveDatasetLabels(Set<Statement> labels, RepositoryConnection con, URI dsURI) throws Exception{

		StringBuilder query = new StringBuilder("SELECT ?title ?desc ");
		query.append("WHERE { <").append(dsURI.stringValue()).append("> <")
			.append(SKOS.PREFLABEL.stringValue()).append("> ?title . ")
		.append(" OPTIONAL { <")
			.append(dsURI.stringValue()).append("> <").append(SKOS.DEFINITION.stringValue()).append("> ?desc . }  } ");

		TupleQuery tquery = con.prepareTupleQuery(QueryLanguage.SPARQL, query.toString());
		TupleQueryResult result = null;
		String title = "";
		try{
			result = tquery.evaluate();
			BindingSet bs;
			Literal label;
			while(result.hasNext()){
				bs = result.next();
				if(bs.hasBinding("title")){
					label = (Literal) bs.getValue("title");
					if("en".contains(label.getLanguage())){
						labels.add(vf.createStatement(dsURI, DCTERMS.TITLE, label));
						title = label.stringValue();
					}
				}
				if(bs.hasBinding("desc")){
					label = (Literal) bs.getValue("desc");
					if("en".contains(label.getLanguage())){
						labels.add(vf.createStatement(dsURI, DCTERMS.DESCRIPTION, label));
					}
				}
			}
		}finally{
			if(result != null){
				result.close();
			}
		}

		setStaticDimensions(labels, dsURI, title);

	}

	private void retrieveEntities(Set<Statement> labels){
		Repository repo = new SPARQLRepository(SPARQL_ENDPOINT);
		RepositoryConnection con = null;
		Set<URI> entities = new  HashSet<URI>();
		try{
			repo.initialize();
			con = repo.getConnection();
		}catch(RepositoryException e){
			logger.warn("Could not init SPARQL endpoint at " + SPARQL_ENDPOINT , e);
		}

		StringBuilder query = new StringBuilder();
		query.append(" SELECT DISTINCT ?dataset ?country ");
		query.append(" FROM NAMED <").append(INDICATOR_GRAPH).append("> ");
		query.append(" WHERE { ");
		query.append(" ?obs rdf:type qb:Observation ; ");
		query.append(" qb:dataSet ?dataset ; ");
		query.append(" <").append(Vocabulary.SDMX.REF_AREA.stringValue()).append("> ?country . } ");


		TupleQueryResult result = null;

		try{
			TupleQuery tquery = con.prepareTupleQuery(QueryLanguage.SPARQL, query.toString());
			try{
				result = tquery.evaluate();

				BindingSet bs;
				while(result.hasNext()){
					bs = result.next();

					labels.add(vf.createStatement((URI) bs.getValue("dataset"), SDMX.REF_AREA, bs.getValue("country")));
					entities.add((URI) bs.getValue("country"));
				}
			} finally{
				if(result != null){
					result.close();
				}
			}
		}catch(RepositoryException e){
			logger.warn("Could not query repository at " + SPARQL_ENDPOINT , e);
		} catch (MalformedQueryException e) {
			logger.warn("Malformed query", e);
		}catch (QueryEvaluationException e) {
			logger.warn("QueryEvaluationException ", e);
		}


		// shutdown
		try{
			if(con != null){
				con.close();
			}
			repo.shutDown();
		}catch(RepositoryException e){
			logger.warn("Could not shutdown SPARQL endpoint at " + SPARQL_ENDPOINT, e);
		}

		getEntityLabels(labels, entities);
	}

	private void getEntityLabels(Set<Statement> labels, Set<URI> entities){

		StringBuilder query = new StringBuilder();
		for(URI entity : entities){

			query.append(" SELECT ?label ");
			query.append(" FROM <").append(META_GRAPH).append("> ");
			query.append(" WHERE { ");
			query.append(" <").append(entity.stringValue()).append("> <").append(SKOS.PREFLABEL).append("> ?label . } ");

			try{
				Repository repo = new SPARQLRepository(SPARQL_ENDPOINT);
				repo.initialize();
				RepositoryConnection con = repo.getConnection();

				TupleQuery tquery = con.prepareTupleQuery(QueryLanguage.SPARQL, query.toString());
				TupleQueryResult result = null; 

				try{
					result = tquery.evaluate();

					BindingSet bs;
					while(result.hasNext()){
						bs = result.next();

						labels.add(vf.createStatement(entity, SKOS.PREFLABEL, bs.getValue("label")));
					}
				}finally{
					if(result != null){
						result.close();
					}
				}

				con.close();
				repo.shutDown();

			}catch(RepositoryException e){
				logger.warn("Could not query repository at " + SPARQL_ENDPOINT , e);
			} catch (MalformedQueryException e) {
				logger.warn("Malformed query", e);
			}catch (QueryEvaluationException e) {
				logger.warn("QueryEvaluationException ", e);
			}
		}
	}


	private void setStaticDimensions(Set<Statement> labels, URI dataset, String title){

		//The country or geographic area to which the measured statistical phenomenon relates.
		labels.add(vf.createStatement(dataset, DataCube.DIMENSION, SDMX.REF_AREA));
		labels.add(vf.createStatement(SDMX.REF_AREA, RDF.TYPE, DataCube.DIMENSION_PROPERTY));
		labels.add(vf.createStatement(SDMX.REF_AREA, SKOS.PREFLABEL, vf.createLiteral("The country or geographic area to which the measured statistical phenomenon relates.")));

		labels.add(vf.createStatement(dataset, DataCube.DIMENSION, SDMX.REF_PERIOD));
		labels.add(vf.createStatement(SDMX.REF_PERIOD, RDF.TYPE, DataCube.DIMENSION_PROPERTY));
		labels.add(vf.createStatement(SDMX.REF_PERIOD, SKOS.PREFLABEL, vf.createLiteral("The period of time or point in time to which the measured observation is intended to refer.")));

		labels.add(vf.createStatement(dataset, DataCube.DIMENSION, vf.createURI("http://worldbank.270a.info/property/", dataset.getLocalName())));
		labels.add(vf.createStatement(vf.createURI("http://worldbank.270a.info/property/", dataset.getLocalName()), RDF.TYPE, DataCube.DIMENSION_PROPERTY));
		labels.add(vf.createStatement(vf.createURI("http://worldbank.270a.info/property/", dataset.getLocalName()), SKOS.PREFLABEL, vf.createLiteral(title)));
	}

	private List<URI> getDatasetURIs(RepositoryConnection con) throws Exception{
		List<URI> ds = new LinkedList<URI>();

		RepositoryResult<Statement> result = null;
		try{
			result = con.getStatements(vf.createURI(DATASET_META_NAMESPACE,"indicator"), SKOS.HAS_TOP_CONCEPT, null, false);
			Statement s;
			while(result.hasNext()){
				s = result.next();
				if(s.getObject() instanceof URI){
					ds.add((URI) s.getObject());
				}
			}
		}finally{
			if(result != null){
				result.close();
			}
		}
		return ds;
	}

	private InputStream establishLink(URL url){
		InputStream is = null;
		try {
			HttpURLConnection httpcon = (HttpURLConnection) url.openConnection();
			httpcon.setRequestMethod("GET");
			if(httpcon.getResponseCode() == HttpURLConnection.HTTP_OK){
				is = httpcon.getInputStream();
			}
		} catch (IOException e) {
			logger.warn("Could not establish connection to " + url.toString(), e);
		}
		return is;
	}
}
